package com.nepu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.nepu.gmall.realtime.bean.TableProcess;
import com.nepu.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * @author chenshuaijun
 * @create 2023-02-16 16:20
 */
public class MyBroadcastFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    // jdbc的连接器
    private Connection connection;
    // MapState的描述器，因为之前我们定义过，所以这里可以把它当做是一个变量，直接传递过来
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public MyBroadcastFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
//        Class.forName(GmallConfig.PHOENIX_DRIVER);
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    // 先处理广播的数据
    // 具体得到的数据结构：{"before":null,"after":{"source_table":"a","sink_table":"b","sink_columns":"c,d,e","sink_pk":"d","sink_extend":"f"},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1676535970761,"snapshot":"false","db":"gmall_config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1676535970767,"transaction":null}
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception{
        PreparedStatement preparedStatement = null;
        String tableName = null;
        try {
            // 1、先将数据转换成JSON对象
            JSONObject jsonObject = JSON.parseObject(value);
            // after这部分的数据是我们需要的数据，所以我们取出这部分的数据
            String after = jsonObject.getString("after");
            // 为了方便使用，我们可以将数据解析成指定的对象
            TableProcess tableProcess = JSON.parseObject(after, TableProcess.class);
            // 这个是为了打印异常而添加的
            tableName = tableProcess.getSinkTable();
            // 拼接对应创建表的语句：create table if not exist table_name(id varchar primary key, name varchar)...
            StringBuilder createTableSql = new StringBuilder("create table if not exists ")
                    .append(GmallConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(tableProcess.getSinkTable())
                    .append("(");

            // 进行字段的拼接
            String sinkColumns = tableProcess.getSinkColumns();
            // 处理对应的sinkPK，因为可能当前的sinkPK是一个null
            String sinkPK = tableProcess.getSinkPK();
            if (sinkPK == null || sinkPK.length() == 0){
                sinkPK = "id";
            }
            String[] columns = sinkColumns.split(",");
            for (int i = 0; i < columns.length; i++) {
                if (columns[i].equals(sinkPK)){
                    createTableSql.append(columns[i]).append(" varchar primary key");
                } else {
                    createTableSql.append(columns[i]).append(" varchar");
                }

                if (i < columns.length-1){
                    createTableSql.append(",");
                } else {
                    createTableSql.append(")");
                }
            }
            if (tableProcess.getSinkExtend() != null){
                createTableSql.append(tableProcess.getSinkExtend());
            }
            String sql = createTableSql.toString();
            System.out.println("拼接的sql是："+sql);
            preparedStatement = connection.prepareStatement(sql);
            // 执行对应的sql
            preparedStatement.execute();

            //TODO 3.写入状态,广播出去：这个是我在自己写的时候，忘记的点，我创建完表之后，没有将状态广播出去，所以导致，后面的主流拿不到对应的广播数据
            BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
            broadcastState.put(tableProcess.getSourceTable(), tableProcess);
        } catch (SQLException e) {
            // 这里如果出现了建表失败，我们直接将整个程序停止了.因为即使这里将对应的异常捕获了，也会在后面将数据插入到表中的时候，报表找不到。
            throw new RuntimeException("创建表失败："+tableName);
        } finally {
            // 关闭对应的连接
            if (preparedStatement != null){
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    // 处理主流中的数据
    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // 首先取出广播数据
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        // 取出主流中当前这条数据是来自于那个表
        String tableName = value.getString("table");
        // 从广播的数据中取出对应的数据
        TableProcess tableProcess = broadcastState.get(tableName);
        // 如果当前这条数据所属的表，并不属于动态表清单中已经配置的表中的数据，他就无法从当前的广播数据中取出数据。所以得到的可能是一个null
        if (tableProcess != null){
            // 如果我们能取到数据，就表明当前的数据是属于表清单中的数据，我们就需要过滤掉不需要的字段
            // 先取出我们需要的数据
            String sinkColumns = tableProcess.getSinkColumns();
            String[] columns = sinkColumns.split(",");
            List<String> columnsList = Arrays.asList(columns);
            Set<Map.Entry<String, Object>> entries = value.getJSONObject("data").entrySet();
            Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, Object> entry = iterator.next();
                // 如果不包括这个key就将对应的key移除
                if (!columnsList.contains(entry.getKey())){
                    iterator.remove();
                }
            }
            // 处理完成之后，加上一个sinkTable的字段
            value.put("sinkTable", tableProcess.getSinkTable());
            // 将数据写出
            out.collect(value);
        }
    }

    // 关闭连接
    @Override
    public void close() throws Exception {
        connection.close();
    }
}
